Thema: Jabber|XMPP ChatGPT Bot (LIVE DEMO!)

Hoi Hoi Szenebox ich wollte euch Heute mal eine kleinere Spielerei von mir vorstellen.
Ich habe die ChatGPT API mit dem davinci-003 Model mit XMPP/Jabber gekoppelt (leider ohne OTR)

Mit "reset" könnt ihre eure aktuelle Account gebundene Instanz zurücksetzen


Wie benutze ich den Bot?
Einfach [email protected] anschreiben (OTR noch in der mache)


Beispiel:

(14:23:32) kickdowncorsa: reset
(14:23:32) [email protected]: Chatverlauf wurde zurückgesetzt.
14:50
(14:53:33) kickdowncorsa: Schreibe eine kurze Vorstellung über einen XMPP Bot der mit Jabber gekoppelt ist DU AI MISTMADE
(14:53:37) [email protected]: Mein Name ist MistMade und ich bin ein XMPP Bot, der mit Jabber gekoppelt ist. Ich kann Ihnen bei verschiedenen Aufgaben helfen, wie zum Beispiel Nachrichten senden und empfangen, Benutzer suchen und hinzufügen sowie an Chatrooms teilnehmen. Ich bin sehr leistungsstark und benutzerfreundlich. Schau dir meine Funktionen an; du wirst sicher keine Enttäuschung erleben!




Source Code der live Demo:

Code:

import asyncio
import openai
from slixmpp import ClientXMPP
import itertools

API_KEYS = [
    "sk-K3RpApS2OEIxnXQjxxXXXXXXXXXXXXXXXXXXXXXXX",
    "sk-1eZQfxCL5dF2Zwm5RubIXXXXXXXXXXXXXXXXXXXXX"
]

api_key_cycle = itertools.cycle(API_KEYS)

JID = "[email protected]"
PASSWORD = "XXXXXXXXXXXXXXXxx"

class ChatGPTXMPP(ClientXMPP):
    def __init__(self, jid, password):
        super().__init__(jid, password)
        self.add_event_handler("session_start", self.start)
        self.add_event_handler("message", self.message)
        self.chat_histories = {}

    async def start(self, event):
        self.send_presence()
        await self.get_roster()

    async def message(self, msg):
        if msg["type"] in ("chat", "normal"):
            user_input = msg["body"]
            user_jid = str(msg["from"])

            # Verlauf zurücksetzen, wenn der Benutzer "reset" sendet
            if user_input.lower() == "reset":
                self.reset_chat_history(user_jid)
                self.send_message(mto=msg["from"], mbody="Chatverlauf wurde zurückgesetzt.", mtype="chat")
                return

            if user_jid not in self.chat_histories:
                self.chat_histories[user_jid] = []
            self.chat_histories[user_jid].append({"role": "User", "content": user_input})

            chatgpt_response = await self.chat_gpt(user_jid, user_input)
            self.chat_histories[user_jid].append({"role": "ChatGPT", "content": chatgpt_response})
            self.send_message(mto=msg["from"], mbody=chatgpt_response, mtype="chat")

    def reset_chat_history(self, user_jid):
        if user_jid in self.chat_histories:
            del self.chat_histories[user_jid]

    async def chat_gpt(self, user_jid, user_input):
        api_key = next(api_key_cycle)
        openai.api_key = api_key
        print(api_key)

        chat_history = self.chat_histories[user_jid]
        prompt = ""
        for message in chat_history:
            prompt += f"{message['role']}: {message['content']}\n"
        prompt += "ChatGPT:"

        try:
            response = await asyncio.to_thread(
                openai.Completion.create,
                engine="text-davinci-003",
                prompt=prompt,
                temperature=0.7,
                max_tokens=1000,
                top_p=1,
                frequency_penalty=1,
                presence_penalty=1,
            )
            return response.choices[0].text.strip()
        except Exception as e:
            print(f"Error: {e}")
            return "Entschuldigung, es gab einen Fehler bei der Verarbeitung Ihrer Anfrage."

if __name__ == "__main__":
    xmpp = ChatGPTXMPP(JID, PASSWORD)
    xmpp.connect()
    xmpp.process(forever=False)